package com.ruoyi.web.controller.nursestation;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.ruoyi.common.constant.StringConstants;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.my.data.SessionParam;
import com.ruoyi.my.tool.SpringUtility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 报警消息实时弹框推送
 */
@ServerEndpoint(value = "/websocket/{dept}/{ward}/{bed}")
@Component
public class MonitoLargeScreen {

    private static final Logger log = LoggerFactory.getLogger(MonitoLargeScreen.class);
    @Resource
    private RedisCache redisCache = SpringUtility.getBean(RedisCache.class);


    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;

    //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。
    private static final CopyOnWriteArraySet<MonitoLargeScreen> webSocketSet = new CopyOnWriteArraySet<>();
    private final Map<String, SessionParam> sessionParams = new HashMap<>();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("dept") int dept, @PathParam("ward") int ward,
                       @PathParam("bed") int bed) {
        this.session = session;
        webSocketSet.add(this);
        addOnlineCount();
        log.info("新连接加入: dept={}, ward={}, bed={}. 当前在线: {}", dept, ward, bed, getOnlineCount());

        // 存储会话参数
        sessionParams.put(session.getId(), new SessionParam(dept, ward, bed));

        // 使用调度线程池替代循环检查
        CompletableFuture.runAsync(() -> monitorDeviceWarnings(session), warningMonitorExecutor);
    }

    // 创建专用线程池（类初始化时创建）
    private static final ExecutorService warningMonitorExecutor = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors(),
            new ThreadFactoryBuilder().setNameFormat("warning-monitor-%d").build()
    );

    // 设备告警监控方法
    private void monitorDeviceWarnings(Session session) {
        Map<String, String> messageTemplate = new HashMap<>(8);
        messageTemplate.put("cmd", "topic");

        while (session.isOpen()) {
            try {
                Map<String, String> devWarnCache = redisCache.getCacheObject("dev_warn_cache");
                if (devWarnCache == null || devWarnCache.isEmpty()) {
                    sleepQuietly(1000);  // 无数据时休眠
                    continue;
                }

                for (Map.Entry<String, String> entry : devWarnCache.entrySet()) {
                    String deviceId = entry.getKey();
                    String warningData = entry.getValue();

                    // 检查是否已发送过告警
                    Set<String> notifiedSessions = getOrCreateNotifiedSet(deviceId);
                    if (notifiedSessions.contains(session.getId())) {
                        continue;
                    }

                    // 准备告警消息
                    prepareWarningMessage(messageTemplate, deviceId, warningData);

                    // 发送并标记为已通知
                    sendMessage(JSON.toJSONString(messageTemplate));
                    markAsNotified(deviceId, session.getId(), notifiedSessions);
                }

                sleepQuietly(500);  // 处理间隔

            } catch (Exception e) {
                log.error("设备告警监控异常: {}", e.getMessage());
                sleepQuietly(1000);  // 异常后休眠
            }
        }
    }

    // 辅助方法
    private Set<String> getOrCreateNotifiedSet(String deviceId) {
        Set<String> notifiedSessions = redisCache.getCacheSet(deviceId);
        return notifiedSessions != null ? notifiedSessions : new CopyOnWriteArraySet<>();
    }

    private void prepareWarningMessage(Map<String, String> template, String deviceId, String warningData) {
        template.put("deviceId", deviceId);

        String[] parts = warningData.split(StringConstants.TRIPLE_PERCENT);
        if (parts.length >= 2) {
            template.put("displayAlarm", parts[1]);
            String msgTxt = parts[0];
            template.put("msgTxt", msgTxt);
            String[] temp = msgTxt.split(StringConstants.UNDERSCORE);
            String[] systemParts = temp[0].split(StringConstants.COLON);
            template.put("deviceName", systemParts[1]);
            template.put("systemId", systemParts[0]);
        }
    }

    private void markAsNotified(String deviceId, String sessionId, Set<String> notifiedSessions) {
        notifiedSessions.add(sessionId);
        redisCache.setCacheSet(deviceId, notifiedSessions);
    }

    private void sleepQuietly(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException ignored) {
            Thread.currentThread().interrupt();
        }
    }

    public void sendMessage(String message) throws Exception {
        if (null != sessionParams.get(this.session.getId())) {
            this.session.getBasicRemote().sendText(message);
            Thread.sleep(1000);
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        sessionParams.remove(this.session.getId());
        webSocketSet.remove(this);  //从set中删除
        subOnlineCount();           //在线数减1
        log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        // System.out.println("session:" + session);
        // System.out.println("来自客户端的消息:" + message);
        // System.out.println(JSON.toJSONString(session.getRequestParameterMap()));
        SessionParam param = JSONObject.parseObject(message, SessionParam.class);
        // 更新 session 携带的参数
        sessionParams.put(session.getId(), param);
    }

    /**
     * 错误时
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.info("【websocket】发生错误：" + error.getMessage());
    }


    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        MonitoLargeScreen.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        MonitoLargeScreen.onlineCount--;
    }
}
